import { Processor, WorkerHost } from "@nestjs/bullmq";
import { Inject, Logger } from "@nestjs/common";
import { ElasticsearchService } from "@nestjs/elasticsearch";
import { InjectDataSource } from "@nestjs/typeorm";
import dayjs from "dayjs";
import { Redis } from "ioredis";
import { DataSource } from "typeorm";
import { DatabaseName } from "@/database/database.enum";
import { REDIS_CLIENT } from "@/database/redis.module";

@Processor("sysLogToEsQueue")
export class SysLogToEsProcessor extends WorkerHost {
    private readonly logger = new Logger(SysLogToEsProcessor.name);
    private readonly REDIS_KEY = "sys_log:es_sync_time";

    constructor(
        @InjectDataSource(DatabaseName.LOGS) private readonly dataSource: DataSource,
        @Inject(REDIS_CLIENT) private readonly redis: Redis,
        private readonly es: ElasticsearchService,
    ) {
        super();
    }

    async process() {
        const batchSize = 1000; // 每次同步的日志数量上限, 根据实际情况调整

        try {
            await this.redis.select(15);
            const storedTime = await this.redis.get(this.REDIS_KEY);
            // toDate()初始时间少8个小时是时区显示问题, 实际的时间戳是相等的
            let lastSyncTime = storedTime
                ? new Date(storedTime)
                : dayjs().subtract(1, "month").startOf("month").toDate();

            let logs: any[] = [];
            let tableName = "";

            // 循环查找有数据的表，或者直到当前时间
            // 限制循环次数，防止阻塞过久（例如每次最多推进 12 个月）
            let iterations = 0;
            const maxIterations = 12;

            while (logs.length === 0 && iterations < maxIterations) {
                iterations++;
                const lastDate = dayjs(lastSyncTime);
                tableName = `sys_operation_log_${lastDate.format("YYYYMM")}`;

                // 检查表是否存在
                const res = await this.dataSource.query(`SELECT to_regclass('${tableName}') as reg`);
                const tableExists = !!res[0]?.reg;

                if (!tableExists) {
                    if (lastDate.isBefore(dayjs(), "month")) {
                        lastSyncTime = lastDate.add(1, "month").startOf("month").toDate();
                        continue; // 表不存在且是过去的时间 -> 跳到下个月
                    } else {
                        await this.redis.set(this.REDIS_KEY, lastSyncTime.toISOString());
                        return; // 当前月的表还不存在 -> 结束
                    }
                }

                // 表存在，查询数据
                logs = await this.dataSource.query(
                    `SELECT id, module, type, action, operator_id, user_agent, ip_address, created_at FROM ${tableName} WHERE created_at > $1 ORDER BY created_at ASC LIMIT $2`,
                    [lastSyncTime, batchSize],
                );

                if (logs.length === 0) {
                    if (lastDate.isBefore(dayjs(), "month")) {
                        // 表存在但无新数据，且是过去的时间 -> 跳到下个月
                        lastSyncTime = lastDate.add(1, "month").startOf("month").toDate();
                    } else {
                        await this.redis.set(this.REDIS_KEY, lastSyncTime.toISOString());
                        return; // 当前月无新数据 -> 结束
                    }
                }
            }

            if (logs.length === 0) {
                await this.redis.set(this.REDIS_KEY, lastSyncTime.toISOString());
                return;
            }

            const operations = new Array(logs.length * 2);
            const len = logs.length;

            for (let i = 0; i < len; i++) {
                const log = logs[i];
                operations[i * 2] = { index: { _index: "sys_logs", _id: log.id } };
                operations[i * 2 + 1] = {
                    id: log.id,
                    module: log.module,
                    type: log.type,
                    action: log.action,
                    operatorId: log.operator_id,
                    ipAddress: log.ip_address,
                    userAgent: log.user_agent,
                    createdAt: log.created_at.toISOString(),
                };
            }

            const result = await this.es.bulk({ operations, refresh: false });

            if (result.errors) {
                const erroredItems = result.items.filter((item: any) => {
                    const operation = Object.keys(item)[0];
                    return item[operation].error;
                });
                this.logger.error(`Bulk insert errors: ${JSON.stringify(erroredItems)}`);
            }

            // 更新增量时间
            lastSyncTime = logs[len - 1].created_at;
            await this.redis.set(this.REDIS_KEY, lastSyncTime.toISOString());

            this.logger.log(
                `Synced ${logs.length} logs from ${tableName}. lastSyncTime = ${lastSyncTime.toISOString()}`,
            );
        } catch (err) {
            this.logger.error("Sync logs failed", err);
        }
    }
}
